package com.liuyu.common.framework.config;

import com.liuyu.common.framework.lang.utils.ApplicationContextUtils;
import com.liuyu.common.framework.rocketmq.properties.ConsumerGroup;
import com.liuyu.common.framework.rocketmq.properties.RocketMqConsumeProperties;
import com.liuyu.common.framework.rocketmq.properties.TopicSubscribe;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @ClassName RocketMqConsumerConf

 * @Author: liuyu
 * @Date: Created in 2021/7/30 15:05
 * @Version: 1
 * @Modified By:
 */
@Component
@Slf4j
public class RocketMqConsumerConf implements SmartInitializingSingleton {

    private static final String CLUSTERING = "CLUSTERING";
    @Autowired
    private RocketMqConsumeProperties consumeProperties;
    private AtomicBoolean flag = new AtomicBoolean(false);

    @Override
    public void afterSingletonsInstantiated() {

        if (flag.compareAndSet(false, true)) {
            log.info("正在创建消费者---------------------------------------");
            List<ConsumerGroup> consumerList = consumeProperties.getConsumerList();
            // 初始化消费者
            if (CollectionUtils.isEmpty(consumerList)) {
                log.info("无MQ消费者---------------------------------------");
            }
            consumerList.forEach(consumer -> {
                // 此步可以针对配置参数进行校验，校验consumer格式，符合规则才启动--待完善

                DefaultMQPushConsumer start = new DefaultMQPushConsumer(consumer.getGroupName());
                start.setNamesrvAddr(consumer.getNamesrvAddr());
                start.setConsumeThreadMin(consumer.getConsumeThreadMin());
                start.setConsumeThreadMax(consumer.getConsumeThreadMax());
                start.setConsumeMessageBatchMaxSize(consumer.getConsumeMessageBatchMaxSize());
                // 设置消费模型，集群(MessageModel.CLUSTERING)还是广播(MessageModel.BROADCASTING)，默认为集群
                String messageModel = consumer.getMessageModel();
                if (StringUtils.isBlank(messageModel) || StringUtils.equals(CLUSTERING, messageModel)) {
                    start.setMessageModel(MessageModel.CLUSTERING);
                } else {
                    start.setMessageModel(MessageModel.BROADCASTING);
                }
                // 设置consumer第一次启动是从队列头部开始还是队列尾部开始,否则按照上次消费的位置继续消费
                start.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
                // 设置监听,判断是否为顺序消费
                Boolean orderly = consumer.getOrderly();
                if (orderly == null || orderly) {
                    // 顺序消费
                    start.registerMessageListener((MessageListenerOrderly) ApplicationContextUtils.getApplicationContext().getBean(consumer.getListenerBeanName()));
                } else {
                    start.registerMessageListener((MessageListenerConcurrently) ApplicationContextUtils.getApplicationContext().getBean(consumer.getListenerBeanName()));
                }

                List<TopicSubscribe> subscribes = consumer.getSubscribes();
                if (CollectionUtils.isEmpty(subscribes)) {
                    // 未配置主题，则不启动
                    return;
                }
                try {
                    for (TopicSubscribe topic : subscribes) {
                        start.subscribe(topic.getTopicName(), StringUtils.isBlank(topic.getTagName()) ? "*" : topic.getTagName());
                    }
                    start.start();
//                startConsumer.add(start);
                    log.info("MQ消费者group:{},namesrv:{}启动成功", consumer.getGroupName(), consumer.getNamesrvAddr());
                } catch (MQClientException e) {
                    log.error("MQ消费者group:" + consumer.getGroupName() + ",namesrv:" + consumer.getNamesrvAddr() + "启动失败", e);
                }
            });
        }
    }
}
